package com.etl.component.basic;

import com.google.common.collect.Lists;
import com.component.api.constants.ComponentCodeConstants;
import com.component.api.constants.ComponentConstants;
import com.component.api.model.PortData;
import com.component.api.model.ProcessResult;
import com.component.api.model.data.ColumnData;
import com.component.api.model.data.SetData;
import com.component.api.model.dto.RuleColumnDTO;
import com.component.api.model.param.ParamPair;
import com.component.api.model.param.TableColumn;
import com.etl.component.common.AbstractFlinkComponent;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.Utils;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;

import java.util.List;
import java.util.Map;

/**
 * 描述：
 * 字段规范
 * @author xianggj
 * @Date 2021/12/3 18:08
 **/
public class DictionarySpecificationComponent extends AbstractFlinkComponent {

    @Override
    public ProcessResult process(List<PortData> datas, List<ParamPair> paramPairs) {
        Map<String, Object> paramMap = paramToObject(paramPairs);
        List<RuleColumnDTO> ruleColumnDTOS =
                (List<RuleColumnDTO>) paramMap.get(ComponentConstants.COLUMNS);
        SetData input = (SetData)datas.get(0).getValue();
        DataSet<Row> dataset = (DataSet<Row>) input.getDataset();
        List<TableColumn> columns = input.getColumns();
        setIndex(ruleColumnDTOS, columns);
        RowTypeInfo type = (RowTypeInfo)dataset.getType();
        DataSet<Row> result = new MapOperator<Row, Row>(dataset, type, e->{
            for (RuleColumnDTO toAcolumnDTO : ruleColumnDTOS) {
                int index = toAcolumnDTO.getIndex();
                Object field = e.getField(index);
                if (field == null){
                    continue;
                }
                String val = field.toString();
                String convert = toAcolumnDTO.getFunction().convert(val);
                e.setField(index, convert);
            }
            return e;
        }, Utils.getCallLocationName());
        return execute(result, columns,"执行字典规范操作");
    }

    @Override
    public List<PortData> previewOutputColumns(List<PortData> datas, List<ParamPair> paramPairs) {
        ColumnData input = (ColumnData)datas.get(0).getValue();
        List<TableColumn> columns = input.getColumns();
        return Lists.newArrayList(new PortData(getRule(), new ColumnData(columns)));
    }

    @Override
    public String getRule() {
        return ComponentCodeConstants.DICTIONARY_SPECIFICATION;
    }
}
